/**
 * Copyright 2018-2118 the original author or authors.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.mykit.db.transfer.sync.impl;

import io.mykit.db.transfer.entity.JobInfo;
import io.mykit.db.transfer.sync.DBSync;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;

/**
 * @author binghe
 * @version 1.0.0
 * @description SQL Server数据库同步实现
 */
public class SQLServerSync extends AbstractDBSync implements DBSync {
    private Logger logger = LoggerFactory.getLogger(SQLServerSync.class);

    @Override
    public String assembleSQL(String srcSql, String destSql, Connection conn, Connection outConn, JobInfo jobInfo) throws SQLException {
        String fieldStr = jobInfo.getDestTableFields();
        String[] fields = jobInfo.getDestTableFields().split(",");
        fields = this.trimArrayItem(fields);
        String destTable = jobInfo.getDestTable();
        // 先查询目标数据库
        Statement pstTarget = outConn.createStatement();
        ResultSet rsTarget = pstTarget.executeQuery(destSql);
        long count = 0;
        StringBuffer sql = new StringBuffer();
        if (rsTarget.next()) {
            // 当目标数据库存在数据时
            // 拼接sql语句
            ResultSetMetaData metaData = rsTarget.getMetaData();
            int columnCount = metaData.getColumnCount();
            String columnName = metaData.getColumnName(columnCount);
            String targetTime = rsTarget.getString(columnName);
            String conditionStr = "where " + columnName + " > '" + targetTime + "'" + " order by " + columnName;
            String exeSql = srcSql.replaceAll(" order by " + columnName, " ") + conditionStr;
            // 查询源数据库满足条件的数据
            Statement stat = conn.createStatement();
            ResultSet rs = stat.executeQuery(exeSql);
            while (rs.next()) {
                sql.append("insert into ").append(destTable).append("(").append(fieldStr).append(") values(");
                for (int index = 0; index < fields.length; index++) {
                    sql.append("'").append(rs.getString(fields[index])).append(index == (fields.length - 1) ? "'" : "',");
                }
                sql.append(");");
                count++;
            }
            this.logger.info("总共查询到 " + count + " 条记录");
            if (rs != null) {
                rs.close();
            }
            if (stat != null) {
                stat.close();
            }
        } else {
            // 当目标数据库没有数据时
            Statement stat = conn.createStatement();
            ResultSet rs = stat.executeQuery(srcSql);
            while (rs.next()) {
                sql.append("insert into ").append(destTable).append("(").append(fieldStr).append(") values(");
                for (int index = 0; index < fields.length; index++) {
                    sql.append("'").append(rs.getString(fields[index])).append(index == (fields.length - 1) ? "'" : "',");
                }
                sql.append(");");
                count++;
            }
            this.logger.info("总共查询到 " + count + " 条记录");
            if (rs != null) {
                rs.close();
            }
            if (stat != null) {
                stat.close();
            }
        }
        if (rsTarget != null) {
            rsTarget.close();
        }
        if (pstTarget != null) {
            pstTarget.close();
        }
        return count > 0 ? sql.toString() : null;
    }

    @Override
    public void executeSQL(String sql, Connection conn) throws SQLException {
        PreparedStatement pst = conn.prepareStatement(sql);
        pst.executeUpdate();
        conn.commit();
        pst.close();
    }
}
